package com.jn.rxjava.views;

import android.annotation.SuppressLint;

import com.jn.rxjava.R;
import com.jn.rxjava.base.BaseActivity;
import com.jn.rxjava.utils.LogUtils;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;
import io.reactivex.rxjava3.observables.GroupedObservable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class MainActivity extends BaseActivity {

    private Observable<String> mObservable;
    private Observable<Integer> mObservable2;
    private Observer<String> mObserver;
    private Observer<String> mObserver2;
    private Observer<Integer> mObserver3;
    private Observer<Integer> mObserver4;
    int z = 1;

    @Override
    public int getLayoutId() {
        return R.layout.activity_main;
    }

    @Override
    public void init() {
        initObservable();
        //initBufferObservable();
        //initGroupByObservable();
        //initFilterObservable();
        //initDistinctObservable();
        //initTakeObservable();
        //initSkipObservable();
        //initElementAtObservable();
        //initTakeLastObservable();
        //initMergeObservable();
        //initConcatObservable();
        //initZipObservable();
        //initCombineLatestObservable();
        /*initObserver();
        mObservable.subscribe(mObserver);
        mObservable.subscribe(mObserver2);
        mObservable2.subscribe(mObserver3);
        mObservable2.subscribe(mObserver4);*/
        //initFlowable();
    }

    private void initFlowable() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> emitter) throws Throwable {
                for (int i = 0; i < 1000; i++) {
                    LogUtils.d("asdf", "========send========" + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.DROP)
                .observeOn(AndroidSchedulers.mainThread())// 指定 最后拿到数据操，解析，显示发生在主线程
                .subscribeOn(Schedulers.computation())// 指定 网络请求耗时操作发生在子线程;
                .subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                LogUtils.d("asdf", "========flow=====onSubscribe===");
                s.request(1000);
            }

            @Override
            public void onNext(Integer s) {
                LogUtils.d("asdf", "========flow========" + s);
            }

            @Override
            public void onError(Throwable t) {
                LogUtils.d("asdf", "========flow=====onError===");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "========flow=====onComplete===");
            }
        });
    }

    /**
     * 创建被观察者
     */
    private void initObservable() {
        //create：手动创建一个 Observable。
        /*mObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                //重写subscribe()定义需要发送的事件
                //ObservableEmitter有三种发射方法，onNext()、onComplete()、onError()
                //onNext()发送事件，可以无限调用，所有的发射信息观察者都可以收得到
                emitter.onNext("电视剧第一集");
                emitter.onNext("电视剧第二集");
                emitter.onNext("电视剧第三集");
                //onComplete()、onError()是互斥的，观察者只会收到一个
                //onComplete()完成事件发送，可以重复调用，但是观察者只会接收一次
                emitter.onComplete();
                //onError()异常发送，不可以重复调用
                //emitter.onError(new IllegalArgumentException("TV Play Exception!"));
            }
        });*/

        //just:创建一个发射指定数据项的 Observable
        mObservable = Observable.just("电视剧第一集", "电视剧第二集", "电视剧第三集");

        //range：创建一个发射特定整数范围的 Observable。
        mObservable2 = Observable.range(1, 10)
                .concatMap(new Function<Integer, ObservableSource<? extends Integer>>() {
                    @Override
                    public ObservableSource<? extends Integer> apply(Integer integer) throws Throwable {
                        List<Integer> list = new ArrayList<>();
                        list.add(integer * 10);
                        list.add(integer * 100);
                        list.add(integer * 1000);
                        return Observable.fromIterable(list);//fromIterable：从一个 Iterable 对象（如 List 或 Set）中创建一个 Observable。
                    }
                });

        //interval： 创建一个 Observable，定期发射一个递增的长整型数值
        /*Observable.interval(1, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Long aLong) {
                LogUtils.d("asdf", "========interval========" + aLong);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });*/


        //timer： 创建一个 Observable，在指定延迟后发射一个数据项
        /*Observable.timer(5, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {//Consumer可理解为简化版的Observer
            @Override
            public void accept(Long aLong) throws Throwable {
                LogUtils.d("asdf", "========timer========" + aLong);
            }
        });*/
    }

    //buffer 用于将数据项分组为列表，并以列表的形式发射
    private void initBufferObservable() {

        Observable.just(1, 2, 3, 4, 5, 6, 7)
                .buffer(2)//每组2条数据
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Throwable {
                        for (int i = 0; i < integers.size(); i++) {
                            LogUtils.d("asdf", z + "=======buffer======get(" + i + ")=" + integers.get(i));
                        }
                        z++;
                    }
                });
    }

    //groupBy ：将 Observable 数据项按某个标准进行分组，然后发射多个子 Observable，每个子 Observable 包含一组具有相同标准的数据项。
    private void initGroupByObservable() {
        Observable.just(-2, -1, 1, 2, 3, 4, 5, 6)
                .groupBy(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Throwable {
                        if (integer > 0) {
                            return "正数";
                        }
                        return "负数";
                    }
                }).subscribe(new Observer<GroupedObservable<String, Integer>>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull GroupedObservable<String, Integer> stringIntegerGroupedObservable) {
                        LogUtils.d("asdf", "========groupBy getkey========" + stringIntegerGroupedObservable.getKey());
                        stringIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Throwable {
                                LogUtils.d("asdf", "========groupBy========" + integer);
                            }
                        });
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //filter：过滤掉不满足条件的数据项。
    @SuppressLint("CheckResult")
    private void initFilterObservable() {
        Observable.just(-2, -1, 1, 2, 3, 4, 5, 6)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer > 0;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========filter========" + integer);
                    }
                });
    }

    //distinct：过滤掉重复的数据项
    //distinctUntilChanged： 用于过滤掉连续重复的数据项，只保留第一次出现的数据项
    @SuppressLint("CheckResult")
    private void initDistinctObservable() {
        Observable.just( 1, 2, 2, 1, 3, 4, 5, 5, 6)
                //.distinct()
                .distinctUntilChanged()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========distinct========" + integer);
                    }
                });
    }

    //take：仅发射前 N 个数据项。
    @SuppressLint("CheckResult")
    private void initTakeObservable() {
        Observable.just( 1, 2, 2, 1, 3, 4, 5, 5, 6)
                .take(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========take========" + integer);
                    }
                });
    }

    //skip：跳过前 N 个数据项。
    @SuppressLint("CheckResult")
    private void initSkipObservable() {
        Observable.just( 1, 2, 2, 1, 3, 4, 5, 5, 6)
                .skip(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========skip========" + integer);
                    }
                });
    }

    //elementAt： 用于发射指定索引位置的数据项，忽略其他数据项。
    @SuppressLint("CheckResult")
    private void initElementAtObservable() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                .elementAt(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========elementAt========" + integer);
                    }
                });
    }

    //takeLast： 用于仅发射最后 N 个数据项，忽略前面的数据项。
    @SuppressLint("CheckResult")
    private void initTakeLastObservable() {
        Observable.just( 1, 2, 2, 1, 3, 4, 5, 5, 6)
                .takeLast(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========takeLast========" + integer);
                    }
                });
    }

    private void initMergeObservable() {
        initObservable();
        Observable.merge(mObservable2, mObservable).subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Object o) {
                LogUtils.d("asdf", "========merge========" + o.toString());
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void initConcatObservable() {
        initObservable();
        Observable.concat(mObservable, mObservable2).subscribeOn(Schedulers.trampoline()).subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Object o) {
                LogUtils.d("asdf", "========concat========" + o.toString());
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void initZipObservable() {
        initObservable();
        Observable.zip(mObservable, mObservable2, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Throwable {
                return s + " " + integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtils.d("asdf", "========zip========" + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void initCombineLatestObservable() {
        Observable<String> oa = Observable.interval(2, TimeUnit.SECONDS)
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Throwable {
                        return aLong + "A";
                    }
                });
        Observable<String> ob = Observable.interval(3, TimeUnit.SECONDS)
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Throwable {
                        return aLong + "B";
                    }
                });
        Observable.combineLatest(oa, ob, new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Throwable {
                return s + " ==== " + s2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                LogUtils.d("asdf", "========combineLatest========" + s);
            }
        });
    }

    /**
     * 创建观察者
     */
    private void initObserver() {
        mObserver = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtils.d("asdf", "onNext : " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "onComplete");
            }
        };

        mObserver2 = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "2 onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtils.d("asdf", "2 onNext : " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "2 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "2 onComplete");
            }
        };

        mObserver3 = new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "3 onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtils.d("asdf", "3 onNext : " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "3 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "3 onComplete");
            }
        };

        mObserver4 = new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "4 onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtils.d("asdf", "4 onNext : " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "4 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "4 onComplete");
            }
        };
    }
}